package zyy;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

class myCallBack implements MqttCallback {
  MQTT service;

  @Override
  public void connectionLost(Throwable cause) {
    System.out.println(/* MQTT.debug_level + */ "当前连接状态\t" + service.client.isConnected());
    // while (true)
    {
      try {
        service.client.disconnectForcibly();
        IMqttToken token = service.client.connectWithResult(service.connOpts);
        System.out.println(token.toString());
      } catch (MqttException e) {
        e.printStackTrace();
        return;
      }
    }
  }

  @Override
  public void deliveryComplete(IMqttDeliveryToken token) {
    System.out.println(/* MQTT.debug_level + */ "消息传送完整性: " + token.isComplete());
  }

  @Override
  public void messageArrived(String topic, MqttMessage message) throws Exception {
    if ("close".equals(message.toString())) {
      // do nothing
    } else {
      System.out.printf(/* MQTT.debug_level + */ "topic: %s \t msg: %s\n", topic, message.toString());
      SaveMysql.savedate(topic, message.getQos(), message.toString());
    }

  }

  public myCallBack setService(MQTT service) {
    this.service = service;
    return this;
  }
}

public class MQTT {
  static String debug_level = "\t\t\t ";
  static final String HOST = "ws://47.93.197.229:8083/mqtt"; /* 服务器地址 */
  String userName = "java_mqtt"; /* 用户名 */
  String password = "123456"; /* 密码 */
  String clientId = "zyy_java_31801146";
  MqttClient client;
  MqttConnectOptions connOpts;
  static ScheduledExecutorService scheduled_pool = Executors.newScheduledThreadPool(20);

  public MQTT(String userName, String password, String clientId) {
    this.userName = userName;
    this.password = password;
    this.clientId = clientId;
    try {
      /* 实例化客户端对象 */
      client = new MqttClient(HOST, clientId, new MemoryPersistence());
      /* 实例化参数配置对象 */
      connOpts = new MqttConnectOptions();
      connOpts.setCleanSession(false);
      connOpts.setUserName(userName);
      connOpts.setPassword(password.toCharArray());
    } catch (MqttException e) {
      e.printStackTrace();
    }
  }

  public void publish(String[] topics, String content, int qos) {
    try {
      /* 建立连接 */
      client.connect(connOpts);

      /* 实例化 MQTT 消息对象，指定内容 */
      MqttMessage message = new MqttMessage(content.getBytes());
      /* 指定 qos 等级 */
      message.setQos(qos);
      /* 发布消息至指定 topics */
      for (int i = 0; i < topics.length; i++) {
        /* 发布至多个主题，只能一个一个发布 */
        client.publish(topics[i], message);
      }

      /* 断开连接 */
      client.disconnect();
    } catch (MqttException e) {
      e.printStackTrace();
    }
  }

  public static void listen(String[] topics, myCallBack callback) {
    synchronized (scheduled_pool) {
      scheduled_pool.schedule(new Thread(() -> {
        try {
          MQTT temp = new MQTT("", "", "");
          /* 建立连接 */
          temp.client.connect();
          /* 可以一次性订阅多个主题 */
          temp.client.subscribe(topics);
          temp.client.setCallback(callback.setService(temp));
        } catch (MqttException e) {
          e.printStackTrace();
        }
      }), 0, TimeUnit.SECONDS);
    }
  }

  public static void stop() {
    synchronized (scheduled_pool) {
      System.out.println("预备终止退出");
      scheduled_pool.shutdownNow();
      Thread.yield();
    }
  }

  public static void main(String[] args) throws InterruptedException {
    MQTT task = new MQTT("java_mqtt", "123456", "zyy_java_31801146");
    task.publish(new String[] { "zyy/1146" }, "nihao woshi Java", 1);
    MQTT.listen(new String[] { "mobile_test", "zyy/1146" }, new myCallBack());
    // MQTT.stop();
  }
}